package com.tcl.kafka.scheduled;

import com.tcl.kafka.config.KafkaStartPauseProperties;
import com.tcl.kafka.constant.StartPauseConstant;
import com.tcl.kafka.core.KafkaService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;

/**
 * 启动定时监听
 * @author liangxi.zeng
 */
@Slf4j
public class StartPauseScheduled {

    private KafkaService kafkaService;

    private KafkaStartPauseProperties kafkaStartPauseProperties;

    public StartPauseScheduled(KafkaService kafkaService, KafkaStartPauseProperties kafkaStartPauseProperties) {
        this.kafkaService = kafkaService;
        this.kafkaStartPauseProperties = kafkaStartPauseProperties;
    }

    @Scheduled(cron = StartPauseConstant.SCHEDULED_START_CRON)
    public void startListener() {
        String listenerIds = kafkaStartPauseProperties.getStartListener();
        log.info("定时启动listener:{}",listenerIds);
        kafkaService.startMultiByComma(listenerIds);
    }

    @Scheduled(cron = StartPauseConstant.SCHEDULED_PAUSE_CRON)
    public void pauseListener() {
        String listenerIds = kafkaStartPauseProperties.getPauseListener();
        log.info("定时暂停listener:{}",listenerIds);
        kafkaService.pauseMultiByComma(listenerIds);
    }
}
